package zju.dgm.service.region;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import zju.dgm.dto.*;
import zju.dgm.service.ServiceEntry;
import zju.dgm.service.dataservice.DataService;
import zju.dgm.service.dataservice.DataUtilProvider;
import zju.dgm.service.distsqlservice.DistSqlUtilProvider;
import zju.dgm.service.election.ElectionService;
import zju.dgm.service.election.MasterUtilProvider;
import zju.dgm.service.name.NameService;
import zju.dgm.util.CacheManager;
import zju.dgm.util.Paths;
import zju.dgm.util.Serializer;
import zju.dgm.util.SyncUtil;

import javax.annotation.Resource;
import java.io.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

@Component
public class RegionService extends ServiceEntry implements RegionUtilProvider {
    public static CountDownLatch finLatch = new CountDownLatch(1);

    // server对应region的表，哈希表存储，用于对内分配server
    private static Map<ServerInfo, List<String>> S2RMap = new ConcurrentHashMap<>();

    // region对应server的表，哈希表存储，用于对外提供接口
    private static Map<String, List<ServerInfo>> R2SMap = new ConcurrentHashMap<>();

    // 当前server数量和冗余度
    private static volatile int serverNum = 0;
    private static volatile int redundancy = 1;

    // 版本控制ZKNode路径
    private String versionNodePath;

    @Resource
    private CuratorFramework client;

    @Resource
    private Paths paths;

    @Resource
    private CacheManager cacheManager;

    @Resource
    protected MasterUtilProvider masterService;

    @Resource
    protected DistSqlUtilProvider disSqlService;

    @Resource
    protected DataUtilProvider dataService;

    private final RegionDataHelper dataHelper = new RegionDataHelper();

    // 日志对象
    private static final Logger logger = LoggerFactory.getLogger(RegionService.class);

    // Master监听线程
    private class MasterListener extends Thread {
        @Override
        public void run() {
            logger.info("分区服务开始监听消息");
            while (true) {
                try {
                    DataFrame req;
                    logger.info("等待接收请求中...");
                    req = dataService.blockingGetOneDF(dataHelper.masterFilter, Integer.MAX_VALUE);
                    logger.info("接收到请求{},开始执行处理", req);
                    ServerInfo sender = SyncUtil.serverInfoMap.get(req.getSender());

                    switch (req.getType()) {
                        default:
                            logger.warn("未知的请求：{}", req);
                            break;
                        case DataFrame.MASTER_MAP_REQ:
                            // 返回Map信息
                            dataService.sendData(sender, dataHelper.mapToData(R2SMap, DataFrame.MASTER_MAP_RES));
                            break;
                        case DataFrame.MASTER_VERSION_REQ:
                            // 返回version和Map信息
                            Stat stat = new Stat();
                            client.getData()
                                    .storingStatIn(stat)
                                    .forPath(versionNodePath);
                            RegionMapInfo<String, ServerInfo> curMapInfo = new RegionMapInfo<>(R2SMap, stat.getVersion());

                            DataFrame data1 = new DataFrame(SyncUtil.serverName, DataFrame.MASTER_VERSION_RES, Serializer.getBytesFromObject(curMapInfo));
                            dataService.sendData(sender, data1);
                            break;
                        case DataFrame.MASTER_FREE_REQ:
                            // 返回分配的serverList
                            DataFrame data2 = dataHelper.listToData(getFreeServers(redundancy, null), DataFrame.MASTER_FREE_RES);
                            dataService.sendData(sender, data2);
                            break;
                        case DataFrame.MASTER_ADD_REQ:
                            // 执行建表操作，更新Map
                            byte[] dataByte = req.getData();
                            AddConfirmInfo<ServerInfo> addInfo = Serializer.getObjectFromBytes(
                                    dataByte,
                                    new MyTR<AddConfirmInfo<ServerInfo>>() {
                                    }
                            );
                            R2SMap.put(addInfo.getName(), addInfo.getList());
                            for (ServerInfo s : addInfo.getList()) {
                                if (!S2RMap.get(s).contains(addInfo.getName())) {
                                    S2RMap.get(s).add(addInfo.getName());
                                }
                            }
                            syncVersion(client, versionNodePath);
                            break;
                        case DataFrame.MASTER_DELETE_REQ:
                            // 执行删表操作，更新Map
                            String name = dataHelper.dataToString(req);
                            List<ServerInfo> list = R2SMap.get(name);
                            R2SMap.remove(name);
                            for (ServerInfo s : list) {
                                S2RMap.get(s).remove(name);
                            }
                            syncVersion(client, versionNodePath);
                            break;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private final MasterListener masterListener = new MasterListener();

    // Server监听线程
    private class ServerListener extends Thread {
        @Override
        public void run() {
            logger.info("分区服务开始监听消息");
            while (true) {
                try {
                    DataFrame req;
                    logger.info("等待接收请求中...");
                    req = dataService.blockingGetOneDF(dataHelper.serverFilter, Integer.MAX_VALUE);
                    logger.info("接收到请求{},开始执行处理", req);
                    ServerInfo sender = SyncUtil.serverInfoMap.get(req.getSender());

                    switch (req.getType()) {
                        case DataFrame.LOG_REQ:
                            // 请求region日志
                            String tableName = dataHelper.dataToString(req);
                            dataService.sendData(sender, disSqlService.getLog(tableName));
                            break;
                        case DataFrame.LOG_INFO:
                            // 请求redo日志
                            disSqlService.reDoWithLog(req);
                            break;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private final ServerListener serverListener = new ServerListener();


    // 打印当前Map信息
    private void printMapInfo() {
        logger.info("[Master Node] Cur serverNum = " + serverNum + " redundancy = " + redundancy);
        logger.info("[Master Node] Cur R2S Map = " + R2SMap);
        logger.info("[Master Node] Cur S2R Map = " + S2RMap);
    }

    // 获取空闲server
    private List<ServerInfo> getFreeServers(int n, List<ServerInfo> curList) {
        List<ServerInfo> allList = new ArrayList<>(S2RMap.keySet());
        if (curList != null)
            allList.removeAll(curList);
        allList.sort(new Comparator<ServerInfo>() {
            @Override
            public int compare(ServerInfo s1, ServerInfo s2) {
                return S2RMap.get(s1).size() - S2RMap.get(s2).size();
            }
        });

        return allList.subList(0, n);
    }

    // 向server中写入region
    private void buildRegion(String name, ServerInfo server) {
        try {
            DataFrame logData = null;
            // 向健在server请求region的log
            if (R2SMap.get(name).contains(SyncUtil.serverInfo)) {
                // 如果本机上有，则直接获取
                logData = disSqlService.getLog(name);
            } else {
                // 如果本机没有，则向其他server请求
                dataService.sendData(R2SMap.get(name).get(0), dataHelper.strToData(name, DataFrame.LOG_REQ));
                logData = dataService.blockingGetOneDF(
                        df -> df != null && df.getSender() != null && df.getData() != null && df.getType() == DataFrame.LOG_INFO
                        , 3000
                );
            }

            // 向新server发送log建表
            dataService.sendData(server, logData);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 更新ZKNode
    private void syncVersion(CuratorFramework client, String path) {
        printMapInfo();
        try {
            // 将R2SMap和S2RMap写入ZK
            MasterMapInfo<String, ServerInfo> mapInfo = new MasterMapInfo<>(R2SMap, S2RMap);
            client.setData().forPath(path, Serializer.getBytesFromObject(mapInfo));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // serverNum变大时，同步冗余度并更新map
    private void syncCurMap() {
        // 冗余度变化时，新分配server并更新map
        int curRedu = (serverNum + 1) / 2;
        if (curRedu != redundancy)
            logger.info("[Master Node] Redundancy changed from " + redundancy + " to " + curRedu);
        for (String name : R2SMap.keySet()) {
            List<ServerInfo> curList = R2SMap.get(name);
            // 如果冗余度不够，则分配
            if (curList.size() < curRedu) {
                List<ServerInfo> newServerList = getFreeServers(curRedu - curList.size(), curList);
                // 更新S2R表
                for (ServerInfo s : newServerList) {
                    buildRegion(name, s);
                    S2RMap.get(s).add(name);
                }
                // 更新R2S表
                R2SMap.get(name).addAll(newServerList);
            }
        }
        // 更新冗余度
        redundancy = curRedu;

        // 更新Map Version
        syncVersion(client, versionNodePath);

    }

    // server挂掉时，分配新server并更新map
    private void syncServer(String serverName) {
        redundancy = (serverNum + 1) / 2;
        ServerInfo downServer = null;
        for (ServerInfo tmpS : S2RMap.keySet()) {
            if (tmpS.getName().equals(serverName)) {
                downServer = tmpS;
                break;
            }
        }
        logger.info("[Master Node] Server down, serverInfo: " + downServer);
        List<String> regionList = S2RMap.get(downServer);
        S2RMap.remove(downServer);

        for (String table : regionList) {
            // 如果冗余度够，则不分配新的server
            R2SMap.get(table).remove(downServer);
            if (R2SMap.get(table).size() >= redundancy) {
                continue;
            }
            // 否则找到新的server
            ServerInfo newServer = getFreeServers(1, R2SMap.get(table)).get(0);
            // 向newServer中写该region
            buildRegion(table, newServer);
            // 更新map
            R2SMap.get(table).add(newServer);
            S2RMap.get(newServer).add(table);
            logger.info("[Master Node] Region " + table + " get new server " + newServer);
        }
        syncVersion(client, versionNodePath);
    }

    @Override
    public String getRootPath() {
        return "/region";
    }

    @Override
    protected CountDownLatch[] getPrerequisites() {
        return new CountDownLatch[]{NameService.finLatch, ElectionService.finLatch, DataService.finLatch};
    }

    @Override
    protected void registerListener() {
        // 只有master才注册监听器
        try {
            masterService.addMasterTask(() -> {
                logger.info("[Master Node] 注册监听器");
                // 监听,在节点挂掉时重新分配
                PathChildrenCache serverInfoCache = cacheManager.getPathChildrenCache(paths.getNamePath());
                serverInfoCache.getListenable().addListener(new PathChildrenCacheListener() {
                    @Override
                    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                        if (event.getData() == null) {
                            return;
                        }
                        String path = event.getData().getPath();
                        if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                            logger.info("[Master Node] server挂掉，开启容错容灾");
                            serverNum = client.getChildren().forPath(paths.getNamePath()).size();
                            syncServer(path.substring(path.lastIndexOf('/') + 1));

                        } else if (event.getType() == PathChildrenCacheEvent.Type.CHILD_UPDATED) {
                            byte[] nodeData = client.getData().forPath(path);
                            if (nodeData != null) {
                                ServerInfo newS = Serializer.getObjectFromBytes(nodeData, ServerInfo.class);
                                if (!S2RMap.containsKey(newS)) {
                                    logger.info("[Master Node] 分区服务监听到新server连入");
                                    if (client.getChildren().forPath(paths.getNamePath()).size() == (serverNum + 1)) {
                                        serverNum++;
                                    }
                                    List<String> newL = Collections.synchronizedList(new ArrayList<>());
                                    S2RMap.put(newS, newL);
                                    syncCurMap();
                                }

                            }
                        }
                    }
                });
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    protected void start() {
        logger.info("分区服务开始");

        try {
            masterService.addMasterTask(() -> {
                // 初始化ZK节点
                versionNodePath = null;
                try {
                    Stat stat = client.checkExists().forPath(getRootPath() + "/version-controller");
                    if (stat == null) {
                        versionNodePath = client.create()
                                .withMode(CreateMode.PERSISTENT)
                                .forPath(getRootPath() + "/version-controller");
                        // 初始化Map信息
                        for (ServerInfo value : SyncUtil.serverInfoMap.values()) {
                            List<String> l = Collections.synchronizedList(new ArrayList<>());
                            S2RMap.put(value, l);
                        }
                    } else {
                        // 从ZK中读取Map信息
                        versionNodePath = getRootPath() + "/version-controller";
                        byte[] nodeData = client.getData().forPath(versionNodePath);
                        if (nodeData == null || nodeData.length == 0) {
                            // 初始化Map信息
                            for (ServerInfo value : SyncUtil.serverInfoMap.values()) {
                                List<String> l = Collections.synchronizedList(new ArrayList<>());
                                S2RMap.put(value, l);
                            }
                        } else {
                            MasterMapInfo<String, ServerInfo> mapInfo = Serializer.getObjectFromBytes(
                                    nodeData,
                                    new MyTR<MasterMapInfo<String, ServerInfo>>() {
                                    }
                            );
                            R2SMap = mapInfo.getR2SMap();
                            S2RMap = mapInfo.getS2RMap();
                            // 同步S2RMap
                            S2RMap.keySet().removeIf(s -> !SyncUtil.serverInfoMap.containsValue(s));
                            if (S2RMap.isEmpty()) {
                                for (ServerInfo value : SyncUtil.serverInfoMap.values()) {
                                    List<String> l = Collections.synchronizedList(new ArrayList<>());
                                    S2RMap.put(value, l);
                                }
                            }
                            // 同步R2SMap
                            if (R2SMap != null) {
                                for (String r : R2SMap.keySet()) {
                                    R2SMap.get(r).removeIf(s -> !SyncUtil.serverInfoMap.containsValue(s));
                                }
                                R2SMap.keySet().removeIf(r -> R2SMap.get(r).isEmpty());
                            }
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }

                logger.info("[Master Node] Map init done");

                // 初始化server数量和当前冗余度
                serverNum = SyncUtil.serverInfoMap.size();
                redundancy = (serverNum + 1) / 2;

                syncCurMap();
                // 注册Master接收函数
                masterListener.start();
                finLatch.countDown();
            });

        } catch (Exception e) {
            e.printStackTrace();
        }

        serverListener.start();


        // 根据是否是master决定服务初始化完成顺序
        if (masterService.masterAvailable()) {
            try {
                finLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } else {
            finLatch.countDown();
        }
        logger.info("分区服务完成");
    }

    @Override
    public Map<String, List<ServerInfo>> getRegionMap() {
        logger.warn("getRegionMap查询,当前masterName:{}，当前serverMap：{}", SyncUtil.masterName, SyncUtil.serverInfoMap);
        if (masterService.masterAvailable()) {
            return R2SMap;
        } else {
            Map<String, List<ServerInfo>> map = null;
            try {
                byte[] emptyByte = null;
                DataFrame data = new DataFrame(SyncUtil.serverInfo.getName(), DataFrame.MASTER_MAP_REQ, emptyByte);
                dataService.sendData(SyncUtil.serverInfoMap.get(SyncUtil.masterName), data);
                DataFrame res = dataService.blockingGetOneDF(
                        df -> df != null && df.getSender() != null && df.getData() != null && df.getType() == DataFrame.MASTER_MAP_RES
                        , 5000
                );
                if (res == null) {
                    return null;
                }
                map = dataHelper.dataToMap(res);
            } catch (Exception e) {
                e.printStackTrace();
            }

            return map;
        }
    }

    @Override
    public RegionMapInfo<String, ServerInfo> checkVersion(int version) {
        RegionMapInfo<String, ServerInfo> newMapInfo = new RegionMapInfo<>();
        if (masterService.masterAvailable()) {
            Stat stat = new Stat();
            try {
                client.getData()
                        .storingStatIn(stat)
                        .forPath(versionNodePath);
            } catch (Exception e) {
                e.printStackTrace();
            }
            newMapInfo.setVersion(stat.getVersion());
            newMapInfo.setMap(R2SMap);
        } else {
            try {
                DataFrame data = new DataFrame(SyncUtil.serverName, DataFrame.MASTER_VERSION_REQ, null);
                dataService.sendData(SyncUtil.serverInfoMap.get(SyncUtil.masterName), data);
                DataFrame res = dataService.blockingGetOneDF(
                        df -> df != null && df.getSender() != null && df.getData() != null && df.getType() == DataFrame.MASTER_VERSION_RES
                        , 3000
                );
                while (res == null) {
                    logger.info("进行一次版本请求重试");
                    dataService.sendData(SyncUtil.serverInfoMap.get(SyncUtil.masterName), data);
                    res = dataService.blockingGetOneDF(
                            df -> df != null && df.getSender() != null && df.getData() != null && df.getType() == DataFrame.MASTER_VERSION_RES
                            , 3000
                    );
                }
                // 解析res数据
                byte[] dataByte = res.getData();
                newMapInfo = Serializer.getObjectFromBytes(
                        dataByte,
                        new MyTR<RegionMapInfo<String, ServerInfo>>() {
                        }
                );
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return newMapInfo;
    }

    @Override
    public List<ServerInfo> addNewRegion(String regionName) {
        if (masterService.masterAvailable()) {
            return getFreeServers(redundancy, null);
        } else {
            List<ServerInfo> resList = null;
            try {
                dataService.sendData(SyncUtil.serverInfoMap.get(SyncUtil.masterName), dataHelper.strToData(regionName, DataFrame.MASTER_FREE_REQ));
                DataFrame res = dataService.blockingGetOneDF(
                        df -> df != null && df.getSender() != null && df.getData() != null && df.getType() == DataFrame.MASTER_FREE_RES
                        , 3000
                );
                resList = dataHelper.dataToList(res);
            } catch (Exception e) {
                e.printStackTrace();
            }

            return resList;
        }
    }

    @Override
    public void addConfirm(String regionName, List<ServerInfo> serverList) {
        if (masterService.masterAvailable()) {
            // 更新R2S表
            R2SMap.put(regionName, serverList);

            // 更新S2R表
            for (ServerInfo s : serverList) {
                if (!S2RMap.get(s).contains(regionName)) {
                    S2RMap.get(s).add(regionName);
                }
            }
            // 更新Map Version
            syncVersion(client, versionNodePath);
        } else {
            try {
                AddConfirmInfo<ServerInfo> addInfo = new AddConfirmInfo<>(regionName, serverList);
                DataFrame data = new DataFrame(SyncUtil.serverInfo.getName(), DataFrame.MASTER_ADD_REQ, Serializer.getBytesFromObject(addInfo));
                dataService.sendData(SyncUtil.serverInfoMap.get(SyncUtil.masterName), data);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void deleteRegion(String regionName) {
        if (masterService.masterAvailable()) {
            // 更新R2S表
            List<ServerInfo> serverList = R2SMap.get(regionName);
            R2SMap.remove(regionName);

            // 更新S2R表
            for (ServerInfo s : serverList) {
                S2RMap.get(s).remove(regionName);
            }

            // 更新Map Version
            syncVersion(client, versionNodePath);
        } else {
            try {
                dataService.sendData(SyncUtil.serverInfoMap.get(SyncUtil.masterName), dataHelper.strToData(regionName, DataFrame.MASTER_DELETE_REQ));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }

}